/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.hadoop.apps.cap3;

import java.io.File;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * Map function execute Cap3 program on a given input file.
 * A list of input file names are passed as the input to the map task
 * using the "Value" object. The file names are separated by a pattern.
 * The map task first, split this string and create a list of file names.
 * Then it calls Cap3 program for each file name, passing it as the input.
 * 
 * 
 * @author Jaliya Ekanayake (jekanaya@cs.indiana.edu)
 * 03/03/2009
 *
 */
public class Cap3Map extends Mapper<Object, Text, IntWritable, Text> {
	
	private String[] getStringArrayFromPatternSeperatedString(String str, String pattern){
			return str.split(pattern);
	}
	
    public void map(Object key, Text value, Context context
    ) throws IOException, InterruptedException {
    	
    	Configuration conf=context.getConfiguration();    	
      	
    	String cap3Dir=conf.get(Cap3Analysis.prop_cap3_dir);	
		String execName=conf.get(Cap3Analysis.prop_exec_name);
		String cmdArgs=conf.get(Cap3Analysis.prop_cmd_args);
		String outputDir=conf.get(Cap3Analysis.prop_output_dir);
		String pattern=conf.get(Cap3Analysis.prop_sep_pattern);	
		
		String files[]=getStringArrayFromPatternSeperatedString(value.toString(),pattern);
		
		for(String fileName :files){
			
			String execCommand =cap3Dir+File.separator+execName+ " "+fileName+" "+ cmdArgs;			
			//We have the full file names in the value. 
			String[] tmp=fileName.split(File.separator);
			String fileNameOnly=tmp[tmp.length-1];//Last part should be the file name.
			
			String stdOutFile=outputDir+File.separator+fileNameOnly+".out";
			String stdErrFile=outputDir+File.separator+fileNameOnly+".err";
			
			try {
				Process p = Runtime.getRuntime().exec(execCommand);
	
				OutputHandler inputStream = new OutputHandler(p.getInputStream(), "INPUT",stdOutFile);
				OutputHandler errorStream = new OutputHandler(p.getErrorStream(), "ERROR",stdErrFile);
	
				// start the stream threads.
				inputStream.start();
				errorStream.start();
	
				p.waitFor();
				
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}
